package com.tt.vtg.scenario.mock;

import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.tt.vtg.scenario.http.pojo.RequestEntity;
import com.tt.vtg.scenario.kafka.KafkaConsumerTask;
import com.tt.vtg.scenario.redis.JedisTemplate;
import com.tt.vtg.scenario.task.CheckOrderStatusTask;
import com.tt.vtg.scenario.util.json.JsonUtils;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class MockCallCentreServer {
    private List<KafkaConsumerTask> kfkComTsks = Lists.newArrayList();
    private Map<String, MockProcessor> processorMap = ImmutableMap.<String, MockProcessor>builder().
            put("INIT", ((orderId, payload) -> {
                JedisTemplate.execute(j ->{
                    if(!j.exists(payload.getUserId())){
                        j.hset(payload.getUserId(), "amount", String.valueOf(payload.getAmount()));
                        j.hset(orderId, "result",
                                payload.getUserId() + " create successful, amount " + payload.getAmount());
                        j.hset(orderId, "status", String.valueOf(CheckOrderStatusTask.SUCCESS));
                    } else {
                        j.hset(orderId, "result",
                                payload.getUserId() + " has already existed. Pls change a userId");
                        j.hset(orderId, "status", String.valueOf(CheckOrderStatusTask.FAILED));
                    }

                    return null;
                });
            })).
            put("ADD", ((orderId, payload) -> {
                JedisTemplate.execute(j ->{
                    if(j.exists(payload.getUserId())){
                        int remain = Integer.valueOf(j.hget(payload.getUserId(), "amount"));
                        j.hset(payload.getUserId(), "amount", String.valueOf(payload.getAmount()+remain));
                        j.hset(orderId, "result",
                                payload.getUserId() +
                                        " add " + payload.getAmount()+ " successful, current amount " + (payload.getAmount()+remain));
                        j.hset(orderId, "status", String.valueOf(CheckOrderStatusTask.SUCCESS));
                    } else {
                        j.hset(orderId, "result",
                                payload.getUserId() + " does not exist!");
                        j.hset(orderId, "status", String.valueOf(CheckOrderStatusTask.FAILED));
                    }

                    return null;
                });
            })).
            put("SUB", ((orderId, payload) -> {
                JedisTemplate.execute(j ->{
                    if(j.exists(payload.getUserId())){
                        int currentAmount = Integer.valueOf(j.hget(payload.getUserId(), "amount"));
                        int remain = currentAmount - payload.getAmount();
                        if(remain >= 0){
                            j.hset(payload.getUserId(), "amount", String.valueOf(remain));
                            j.hset(orderId, "result",
                                    payload.getUserId() +
                                            " sub " + payload.getAmount()+ " successful, current amount " + remain);
                            j.hset(orderId, "status", String.valueOf(CheckOrderStatusTask.SUCCESS));
                        } else {
                            j.hset(orderId, "result",
                                    payload.getUserId() +
                                            " sub " + payload.getAmount()+ " failed, current amount only has " + currentAmount);
                            j.hset(orderId, "status", String.valueOf(CheckOrderStatusTask.FAILED));
                        }
                    } else {
                        j.hset(orderId, "result",
                                payload.getUserId() + " does not exist!");
                        j.hset(orderId, "status", String.valueOf(CheckOrderStatusTask.FAILED));
                    }

                    return null;
                });
            })).build();


    private void run(){
        ExecutorService pool = Executors.newFixedThreadPool(2);

        for(int i=0;i++<2;){
            KafkaConsumerTask task = new KafkaConsumerTask(r->{
                String orderId = r.key();
                RequestEntity payload = JsonUtils.toObj(r.value(), RequestEntity.class);
                process(orderId, payload);
            });
            kfkComTsks.add(task);
            pool.submit(task);
        }

        pool.shutdown();
    }

    private void process(String orderId, RequestEntity payload) {
        if(payload == null || orderId == null){
            return;
        }
        processorMap.get(payload.getAction().name()).execute(orderId, payload);
    }

    private void stop() {
        kfkComTsks.forEach(task -> {
            task.stop();
        });
    }

    public static void main(String[] args) {
        MockCallCentreServer mockServer = new MockCallCentreServer();
        mockServer.run();
        System.out.println("Call center is running...");
    }
}
